﻿using Microsoft.Extensions.DependencyInjection;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using SBlogCore.EventBus.CrossProcessEventBus.Model;
using System.Text;
using Newtonsoft.Json;
using SBlogCore.EventBus.CrossProcessEventBus.MQClient;

namespace SBlogCore.EventBus.CrossProcessEventBus
{
    public class EventBusHandler : IEventBus, IDisposable
    {
        private IModel? _channel;
        private readonly RabbitMqConnection _mQConnection;
        private readonly IConnection _conn;
        private readonly string? _queueName;
        readonly string _exchangeName = "SBlogExchange"; //交换机
        private readonly IServiceScopeFactory _scopeFactory;
        public EventBusHandler(IServiceScopeFactory scopeFactory, string queue, RabbitMqConnection mQConnection)
        {
            this._mQConnection = mQConnection;
            this._conn = mQConnection.Conn();
            this._queueName = queue;
            this._scopeFactory = scopeFactory;
        }
        /// <summary>
        /// 事件集
        /// </summary>
        private Dictionary<string, HashSet<EventMember>> _eventMembers = new();
        /// <summary>
        /// 推送消息
        /// </summary>
        /// <param name="eventName">RoutingKey</param>
        /// <param name="eventData">消息数据</param>
        public Task Publish(string eventName, string eventData)
        {
            using (var model = _conn.CreateModel())
            {
                var properties = model.CreateBasicProperties();
                properties.DeliveryMode = 2;
                byte[] body = Encoding.UTF8.GetBytes(eventData);
                model.BasicPublish(exchange: _exchangeName, routingKey: eventName, mandatory: true, basicProperties: properties, body: body);
            }
            return Task.CompletedTask;
        }
        /// <summary>
        /// 推送消息
        /// </summary>
        /// <param name="eventName"></param>
        /// <param name="eventData"></param>
        /// <returns></returns>
        public Task Publish(string eventName, object?[]? eventData)
        {
            using (var model = _conn.CreateModel())
            {
                var properties = model.CreateBasicProperties();
                properties.DeliveryMode = 2;
                byte[] body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(eventData));
                model.BasicPublish(exchange: _exchangeName, routingKey: eventName, mandatory: true, basicProperties: properties, body: body);
            }
            return Task.CompletedTask;
        }

        /// <summary>
        /// 订阅消息
        /// </summary>
        /// <param name="eventMembers"></param>
        public void Subscribe(Dictionary<string, HashSet<EventMember>> eventMembers)
        {
            _eventMembers = eventMembers;
            if (_channel == null)
            {
                CreateChannel();
            }
            foreach (var member in eventMembers)
            {
                _channel.QueueBind(queue: _queueName, exchange: _exchangeName, routingKey: member.Key);//将队列绑定到交换机上，通过RoutingKey把消息发送到指定的队列
                var consumer = new AsyncEventingBasicConsumer(_channel);//创建接收消息的对象
                consumer.Received += SubscribeReceived;//回调方法
                _channel.BasicConsume(queue: _queueName, autoAck: false, consumer: consumer);
            }
        }
        /// <summary>
        /// 回调函数执行方法
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="args"></param>
        /// <returns></returns>
        private Task SubscribeReceived(object sender, BasicDeliverEventArgs args)
        {
            var eventName = args.RoutingKey;
            var memberList = _eventMembers[eventName];
            var bytes = args.Body.ToArray();
            var msg = Encoding.UTF8.GetString(bytes);
            var tasks = new HashSet<Task>();
            foreach (var member in memberList)
            {
                tasks.Add(Task.Run(() =>
                {
                    using var scope = _scopeFactory.CreateScope();
                    var type = scope.ServiceProvider.GetService(member.InType);
                    var method = member.Method;
                    //获取参数集
                    var pars = method.GetParameters();
                    var jsonArray = JArray.Parse(msg);
                    //如果参数数量不一致，退出
                    if (jsonArray.Count != pars.Length) return;
                    var argsArray = new object?[pars.Length];
                    for (var i = 0; i < jsonArray.Count; i++)
                    {
                        var pType = pars[i].ParameterType;
                        argsArray[i] = jsonArray[i].ToObject(pType);
                    }
                    //判断方法时候为异步方法，如果是异步方法则需要等待
                    var returnType = method.ReturnType;
                    if (returnType == typeof(Task) || returnType.IsSubclassOf(typeof(Task)))
                    {
                        var res= (Task)method.Invoke(type, argsArray);
                        res?.GetAwaiter().GetResult();
                    }
                    else
                    {
                        method.Invoke(type, argsArray);
                    }
                }));
            }
            Task.WaitAll(tasks.ToArray());
            _channel?.BasicAck(args.DeliveryTag, multiple: false);//告诉服务已经消费完了
            return Task.CompletedTask;
        }

        private void CreateChannel()
        {
            _channel = _conn.CreateModel();
            // exchange:交换机名称,
            //type:direct 直连类型、anout 扇型交换机 、topic 主题交换机 、Headers 头部交换机,  
            //durable:交换机是否持久化,将交换机的属性在服务器内部保存，当MQ的服务器发生意外或关闭之后，重启RabbitMQ时不需要重新手动或执行代码去建立交换机，交换机会自动建立，相当于一直存在。
            //autoDelete:如果没有与之绑定的队列是否自动删除交换机
            _channel.ExchangeDeclare(exchange: _exchangeName, type: "direct");//声明交换机
            _channel.QueueDeclare(queue: _queueName, durable: true);//声明一个队列
        }
        //direct:将消息队列绑定到交换机上，当消息到达交换机的时候会携带一个routingKey,交换机会根据routingKey找到指定的队列来消费
        public void Dispose()
        {
            _mQConnection.Clear();
            _channel?.Dispose();
        }
    }
}
